package com.meng.spike.service.impl;


import com.fasterxml.jackson.databind.JavaType;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.meng.spike.entity.GoodsStore;
import com.meng.spike.mapper.GoodsStoreMapper;
import com.meng.spike.repository.GoodsStoreRespository;
import com.meng.spike.service.GoodsStoreService;
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.IOException;
import java.lang.management.ManagementFactory;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.TimeUnit;


@Service
public class MqService {
    private static final Logger LOGGER = LoggerFactory.getLogger(MqService.class);




    public void receive(Map<String,String> message,Channel channel) {
        LOGGER.info("消费内容为：{}", message.get("code")+message.get("count"));
    }

    @Autowired
    private GoodsStoreRespository goodsStoreRespository;


    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Resource(name = "del")
    private DefaultRedisScript<Boolean> redisScriptDel;

    @Resource
    private GoodsStoreService goodsStoreService;

    @Resource
    private RabbitTemplate rabbitTemplate;
    @Resource
    private GoodsStoreMapper goodsStoreMapper;
    /**
     * 消费抢课消息
     * @param
     * @return
     */
    @RabbitListener(queues = "${mq.queue}")
    public void updateGoodsStore(Message message, Channel channel,@Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws IOException {


        try {
            //将JSON格式数据转换为Map对象
            ObjectMapper mapper = new ObjectMapper();
            JavaType javaType = mapper.getTypeFactory().constructMapType(Map.class, String.class, Object.class);
            Map<String, Object> resultMap = mapper.readValue(message.getBody(), javaType);
            String code = (String) resultMap.get("code");
            Integer count = (Integer) resultMap.get("count");
            Integer userId = (Integer) resultMap.get("userId");
            //上锁,用UUID作为特殊标识
            String clientID = UUID.randomUUID().toString();
            try{
                //设置value值为我们生存的clientID
                Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(code,clientID,10, TimeUnit.SECONDS);
                if (!result){
                    //获取锁失败的操作
                    rabbitTemplate.convertAndSend("queue_reply", "抢课人数太多，请稍后再试. _"+userId);
                    channel.basicNack(deliveryTag,true,true);
                    return ;
                }

                //具体业务逻辑...
                GoodsStore goodsStore = goodsStoreMapper.getById(code);
                if(goodsStore != null){
                    if(goodsStore.getStore() <= 0){
                        rabbitTemplate.convertAndSend("queue_reply", "课程选定人数已满，请选择其他课程 _"+userId);

                    }
//                    if(goodsStore.getStore() < count){
//                        rabbitTemplate.convertAndSend("queue_reply", "对不起，库存不足 _"+userId);
//
//                    }
                    System.out.println("剩余库存："+goodsStore.getStore());
                    System.out.println("扣除库存："+count);
                    int i = goodsStoreMapper.update(code, count);
                    if(i>=0){
                        rabbitTemplate.convertAndSend("queue_reply", "选课成功 _"+userId);
                    }else {
                        rabbitTemplate.convertAndSend("queue_reply", "选课失败 _"+userId);
                    }
                }else {
                    rabbitTemplate.convertAndSend("queue_reply", "获取库存失败 _"+userId);
                }
            }finally {
                //lua脚本保证原子性
                List<String> keys = Arrays.asList(code);
                stringRedisTemplate.execute(redisScriptDel, keys, clientID);
            }
            //确认消息
            channel.basicAck(deliveryTag, true);
        } catch (Exception e) {
            e.printStackTrace();
            //拒绝消息
            channel.basicNack(deliveryTag,true,true);
//            channel.basicReject(deliveryTag, true);
        }



    }

    /**
     * 消费回调消息
     * @param
     * @return
     */
    @RabbitListener(queues = "queue_reply")
    public void consumerReply(String message,Channel channel,@Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag){
        //确认消息
        try {
            LOGGER.info(message);
            channel.basicAck(deliveryTag, true);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

}
